package org.example.window;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.example.flink.pojo.WaterSensor;
import org.example.function.WaterSensorFunction;

/**
 * @Author: tang
 * @Description: 增量聚合函数 AggregateFunction
 * @Date 2025/2/19 15:53
 */
public class WindowAggregateApiDemo01 {

    public static void main(String[] args) throws Exception{

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> streamSource = environment.socketTextStream("127.0.0.1", 7777);

        KeyedStream<WaterSensor, String> keyedStream = streamSource.map(new WaterSensorFunction())
                .keyBy(WaterSensor::getId);

        keyedStream
                // 时间窗口
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .aggregate(new AggregateFunction<WaterSensor, Integer, String>() {
                    @Override
                    public Integer createAccumulator() {
                        System.out.println("创建累加器");
                        return 0;
                    }

                    @Override
                    public Integer add(WaterSensor value, Integer accumulator) {
                        System.out.println("add 中间计算 value: --->" + value);
                        return value.getVc() + accumulator;
                    }

                    // 窗口结束会调用的方法
                    @Override
                    public String getResult(Integer accumulator) {
                        return accumulator.toString();
                    }

                    /**
                     * 只有session才会调用
                     * @param a An accumulator to merge
                     * @param b Another accumulator to merge
                     * @return
                     */
                    @Override
                    public Integer merge(Integer a, Integer b) {
                        return null;
                    }
                })
                .print()
        ;

        environment.execute();
    }

}
